package com.explorex.infra.puma.gateway.netty.common.ack;


import com.explorex.infra.puma.gateway.netty.common.BaseSocketSession;
import com.explorex.infra.puma.gateway.netty.common.WaitingAckMessage;
import io.netty.channel.ChannelFutureListener;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;

import java.nio.channels.ClosedChannelException;

/**
 * @author: JacceYang
 * @createTime: 2022/10/21 14:52
 * @description:
 */
public class SendMessageTask implements TimerTask {

    public BaseSocketSession baseSocketSession;

    public WaitingAckMessage waitingAckMessage;
    private long delayTime;

    private int retryTimes;

    private short qos;

    public short getQos() {
        return qos;
    }

    public SendMessageTask(BaseSocketSession baseSocketSession, WaitingAckMessage waitingAckMessage, long delayTime, int retryTimes, short qos) {
        this.baseSocketSession = baseSocketSession;
        this.waitingAckMessage = waitingAckMessage;
        this.delayTime = delayTime;
        this.retryTimes=retryTimes;
        this.qos=qos;
    }


    /**
     * 走到这里来说明 任务重试了,此task 也自动为执行完了
     *
     * @param timeout
     * @throws Exception
     */
    @Override
    public void run(Timeout timeout) throws Exception {
        // 定时任务走了三次以上，取消
        if (retryTimes==0) {
            AckMessageManager.delete(baseSocketSession, waitingAckMessage, AckStatusEnum.TIMES_EXHAUST_FAIL);
            return;
        }
        // 重新写任务,通道不可写了也要取消任务
        if (!baseSocketSession.canWrite()) {
            AckMessageManager.delete(baseSocketSession, waitingAckMessage, AckStatusEnum.SOCKET_CLOSE_FAIL);
            return;
        }

        // 如果已经ack了，那么也取消本次的任务
        if (waitingAckMessage.getStatus() == AckStatusEnum.ACK.getValue()) {
            return;
        }

        // 暂时熔断的消息不做重试，降低对client 端的压力
        if (baseSocketSession.isAckRetryFuse()) {
            AckMessageManager.delete(baseSocketSession, waitingAckMessage, AckStatusEnum.ACK_FUSEED);
            return;
        }

        // 可写了，写一次
        waitingAckMessage.setTryTimes((byte) (waitingAckMessage.getTryTimes() + 1));
        baseSocketSession.sendMessage(waitingAckMessage.getMsg()).addListener((ChannelFutureListener) future -> {
            if (!future.isSuccess() && future.cause() instanceof ClosedChannelException){
                AckMessageManager.delete(baseSocketSession, waitingAckMessage, AckStatusEnum.SOCKET_CLOSE_FAIL);
            }else {
                AckMessageManager.sendMessage(baseSocketSession, waitingAckMessage, delayTime,retryTimes-1,this.getQos());
            }
        });
    }
}
